package com.xwj.KafkaSpark

import java.util.Properties

import org.apache.kafka.clients.producer.{KafkaProducer, ProducerRecord}

import scala.util.Random

/**
  * Created by thinkjoy on 2017/8/10.
  * 生产者
  */
object KafkaSparkProducer {

  def main(args: Array[String]): Unit = {

    val TOPIC="letter"
    val props=new Properties()
    props.put("bootstrap.servers","192.168.0.78:9092")
    props.put("zookeeper.connect","192.168.0.78:2181,192.168.0.75:2181,192.168.0.76:2181")
    props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer")
    props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer")
    props.put("metadata.broker.list", "192.168.0.78:9092,192.168.0.75:9092,192.168.0.76:9092")
    val producter=new KafkaProducer[String,String](props)
    while (true){
      (1 to 100).foreach{messagenum =>
        val str="this is kafka and spark,you know"
        val message=new ProducerRecord[String,String](TOPIC,null,str)
        producter.send(message)
        println("向kafka集群发松送的消息:"+message)
      }
      Thread.sleep(1000)
    }
  }
}
